I'll help you create an MCP server that uses the ras-commander library to query HEC-RAS project information. Based on the notebook, I can see the key functionality you need.Now let me create the supporting files for the MCP server:I've created a complete MCP server implementation for querying HEC-RAS projects using the ras-commander library. Here's what I've built:


#!/usr/bin/env python3
"""
HEC-RAS MCP Server

An MCP server that provides tools for querying HEC-RAS project information
using the ras-commander library.
"""

import asyncio
import json
import logging
from pathlib import Path
from typing import Any, Sequence
import pandas as pd
import io

from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, ToolNotFoundError

# Import ras-commander
try:
    from ras_commander import init_ras_project
except ImportError:
    raise ImportError("ras-commander is not installed. Please install it with: pip install ras-commander")

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize the MCP server
server = Server("hecras-mcp-server")

def dataframe_to_text(df: pd.DataFrame, name: str) -> str:
    """Convert a pandas DataFrame to a formatted text string."""
    if df is None or df.empty:
        return f"\n{name}: No data available\n"
    
    # Use StringIO to capture the DataFrame string representation
    buffer = io.StringIO()
    df.to_string(buf=buffer, max_rows=100, max_cols=None)
    return f"\n{name}:\n{buffer.getvalue()}\n"

@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    """List available tools."""
    return [
        Tool(
            name="query_hecras_project",
            description="Query a HEC-RAS project and return information about plans, geometries, flows, and boundaries",
            inputSchema={
                "type": "object",
                "properties": {
                    "project_path": {
                        "type": "string",
                        "description": "Full path to the HEC-RAS project folder"
                    },
                    "ras_version": {
                        "type": "string",
                        "description": "HEC-RAS version (e.g., '6.5', '6.6')",
                        "default": "6.6"
                    },
                    "include_boundaries": {
                        "type": "boolean",
                        "description": "Include boundary conditions data (can be large)",
                        "default": False
                    }
                },
                "required": ["project_path"]
            }
        ),
        Tool(
            name="get_hecras_plans",
            description="Get only the plans information from a HEC-RAS project",
            inputSchema={
                "type": "object",
                "properties": {
                    "project_path": {
                        "type": "string",
                        "description": "Full path to the HEC-RAS project folder"
                    },
                    "ras_version": {
                        "type": "string",
                        "description": "HEC-RAS version (e.g., '6.5', '6.6')",
                        "default": "6.6"
                    }
                },
                "required": ["project_path"]
            }
        ),
        Tool(
            name="get_hecras_geometries",
            description="Get only the geometries information from a HEC-RAS project",
            inputSchema={
                "type": "object",
                "properties": {
                    "project_path": {
                        "type": "string",
                        "description": "Full path to the HEC-RAS project folder"
                    },
                    "ras_version": {
                        "type": "string",
                        "description": "HEC-RAS version (e.g., '6.5', '6.6')",
                        "default": "6.6"
                    }
                },
                "required": ["project_path"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: Any) -> Sequence[TextContent]:
    """Handle tool calls."""
    
    if name == "query_hecras_project":
        try:
            project_path = Path(arguments["project_path"])
            ras_version = arguments.get("ras_version", "6.6")
            include_boundaries = arguments.get("include_boundaries", False)
            
            # Validate project path
            if not project_path.exists() or not project_path.is_dir():
                return [TextContent(
                    type="text",
                    text=f"Error: The specified project folder does not exist or is not a directory: {project_path}"
                )]
            
            # Initialize the RAS project
            logger.info(f"Initializing HEC-RAS project at: {project_path}")
            ras = init_ras_project(project_path, ras_version)
            
            # Build the response
            response_parts = [
                f"HEC-RAS Project: {ras.project_name}",
                f"Project Path: {project_path}",
                f"HEC-RAS Version: {ras_version}",
                "=" * 80
            ]
            
            # Add plan information
            if hasattr(ras, 'plan_df') and ras.plan_df is not None:
                response_parts.append(dataframe_to_text(ras.plan_df, "PLANS"))
            
            # Add geometry information
            if hasattr(ras, 'geom_df') and ras.geom_df is not None:
                response_parts.append(dataframe_to_text(ras.geom_df, "GEOMETRIES"))
            
            # Add flow information
            if hasattr(ras, 'flow_df') and ras.flow_df is not None:
                response_parts.append(dataframe_to_text(ras.flow_df, "STEADY FLOWS"))
                
            if hasattr(ras, 'unsteady_df') and ras.unsteady_df is not None:
                response_parts.append(dataframe_to_text(ras.unsteady_df, "UNSTEADY FLOWS"))
            
            # Add boundary conditions if requested
            if include_boundaries and hasattr(ras, 'boundaries_df') and ras.boundaries_df is not None:
                response_parts.append(dataframe_to_text(ras.boundaries_df, "BOUNDARY CONDITIONS"))
            
            return [TextContent(
                type="text",
                text="\n".join(response_parts)
            )]
            
        except Exception as e:
            logger.error(f"Error querying HEC-RAS project: {str(e)}")
            return [TextContent(
                type="text",
                text=f"Error querying HEC-RAS project: {str(e)}\n\nPlease ensure:\n1. The project path is correct\n2. HEC-RAS version is installed at the expected location\n3. The project files are valid"
            )]
    
    elif name == "get_hecras_plans":
        try:
            project_path = Path(arguments["project_path"])
            ras_version = arguments.get("ras_version", "6.6")
            
            if not project_path.exists() or not project_path.is_dir():
                return [TextContent(
                    type="text",
                    text=f"Error: The specified project folder does not exist: {project_path}"
                )]
            
            ras = init_ras_project(project_path, ras_version)
            
            response_parts = [
                f"HEC-RAS Project: {ras.project_name}",
                f"Project Path: {project_path}",
                "=" * 80
            ]
            
            if hasattr(ras, 'plan_df') and ras.plan_df is not None:
                response_parts.append(dataframe_to_text(ras.plan_df, "PLANS"))
            else:
                response_parts.append("\nNo plan data available")
            
            return [TextContent(
                type="text",
                text="\n".join(response_parts)
            )]
            
        except Exception as e:
            logger.error(f"Error getting plans: {str(e)}")
            return [TextContent(
                type="text",
                text=f"Error getting HEC-RAS plans: {str(e)}"
            )]
    
    elif name == "get_hecras_geometries":
        try:
            project_path = Path(arguments["project_path"])
            ras_version = arguments.get("ras_version", "6.6")
            
            if not project_path.exists() or not project_path.is_dir():
                return [TextContent(
                    type="text",
                    text=f"Error: The specified project folder does not exist: {project_path}"
                )]
            
            ras = init_ras_project(project_path, ras_version)
            
            response_parts = [
                f"HEC-RAS Project: {ras.project_name}",
                f"Project Path: {project_path}",
                "=" * 80
            ]
            
            if hasattr(ras, 'geom_df') and ras.geom_df is not None:
                response_parts.append(dataframe_to_text(ras.geom_df, "GEOMETRIES"))
            else:
                response_parts.append("\nNo geometry data available")
            
            return [TextContent(
                type="text",
                text="\n".join(response_parts)
            )]
            
        except Exception as e:
            logger.error(f"Error getting geometries: {str(e)}")
            return [TextContent(
                type="text",
                text=f"Error getting HEC-RAS geometries: {str(e)}"
            )]
    
    else:
        raise ToolNotFoundError(f"Unknown tool: {name}")

async def main():
    """Main entry point for the MCP server."""
    async with stdio_server() as (read_stream, write_stream):
        logger.info("Starting HEC-RAS MCP Server...")
        
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="hecras-mcp-server",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == "__main__":
    asyncio.run(main())






#!/usr/bin/env python3
"""
Example client for testing the HEC-RAS MCP Server
"""

import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    """Example of using the HEC-RAS MCP server"""
    
    # Create server parameters - adjust the path to your server.py
    server_params = StdioServerParameters(
        command="python",
        args=["./server.py"],
        env=None
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()
            
            # List available tools
            tools = await session.list_tools()
            print("Available tools:")
            for tool in tools:
                print(f"  - {tool.name}: {tool.description}")
            print()
            
            # Example 1: Query comprehensive project information
            print("Example 1: Querying comprehensive HEC-RAS project information...")
            result = await session.call_tool(
                "query_hecras_project",
                arguments={
                    "project_path": r"D:\CLB-NetworkDrive\25-001 HCFCD Standard Benefits Framework\Sharepoint\South Belt Stormwater Detention Basin (A520-03-00-E003)\8_21_2014\2014.08_Comment_Responses_HEC-RAS_models\2014.08_A100+A120\A120-00-00",
                    "ras_version": "6.6",
                    "include_boundaries": False
                }
            )
            print("Result:")
            print(result.content[0].text)
            print("-" * 80)
            
            # Example 2: Get only plans
            print("\nExample 2: Getting only plans information...")
            result = await session.call_tool(
                "get_hecras_plans",
                arguments={
                    "project_path": r"D:\CLB-NetworkDrive\25-001 HCFCD Standard Benefits Framework\Sharepoint\South Belt Stormwater Detention Basin (A520-03-00-E003)\8_21_2014\2014.08_Comment_Responses_HEC-RAS_models\2014.08_A100+A120\A120-00-00",
                    "ras_version": "6.6"
                }
            )
            print("Plans Result:")
            print(result.content[0].text[:500] + "...")  # Show first 500 chars
            print("-" * 80)
            
            # Example 3: Get only geometries
            print("\nExample 3: Getting only geometries information...")
            result = await session.call_tool(
                "get_hecras_geometries",
                arguments={
                    "project_path": r"D:\CLB-NetworkDrive\25-001 HCFCD Standard Benefits Framework\Sharepoint\South Belt Stormwater Detention Basin (A520-03-00-E003)\8_21_2014\2014.08_Comment_Responses_HEC-RAS_models\2014.08_A100+A120\A120-00-00",
                    "ras_version": "6.6"
                }
            )
            print("Geometries Result:")
            print(result.content[0].text[:500] + "...")  # Show first 500 chars

if __name__ == "__main__":
    asyncio.run(main())











## Created Files:

1. **`server.py`** - The main MCP server implementation with three tools:
   - `query_hecras_project` - Returns comprehensive project information
   - `get_hecras_plans` - Returns only plan information
   - `get_hecras_geometries` - Returns only geometry information

2. **`package.json`** - MCP server configuration for Claude Desktop

3. **`README.md`** - Complete documentation and setup instructions

4. **`requirements.txt`** - Python dependencies

5. **`example_client.py`** - Example client code for testing

## Key Features:

- **Flexible querying**: Can get all project info or specific components
- **Version support**: Configurable HEC-RAS version (6.5, 6.6, etc.)
- **Optional boundaries**: Can include/exclude boundary conditions (which can be large)
- **Error handling**: Validates paths and provides helpful error messages
- **Text output**: Converts pandas DataFrames to readable text format

## Setup Instructions:

1. **Install dependencies**:
   ```bash
   pip install mcp ras-commander pandas
   ```

2. **Configure Claude Desktop** by adding to `claude_desktop_config.json`:
   ```json
   {
     "mcpServers": {
       "hecras": {
         "command": "python",
         "args": ["/path/to/server.py"]
       }
     }
   }
   ```

3. **Usage in Claude**: Once configured, you can ask Claude to query your HEC-RAS projects:
   - "Query the HEC-RAS project at C:/MyProject"
   - "Show me the plans in my HEC-RAS project"
   - "Get the geometries from the river model"

The server will return the DataFrames as formatted text, making it easy to analyze HEC-RAS project structure and components directly within Claude Desktop.


Use the anaconda3 installation in the windows user folder to test the MCP server and add it to the current claude code installation. 

The folder "Muncie" is provided to test the MCP server.


Build out a repository and test the MCP server

